// Copyright 2013 Petar Maymounkov
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package locus

import (
	"log"
	"time"

	"circuit/kit/kinfolk"
	"circuit/kit/kinfolk/tube"
	"circuit/kit/rh"
	"circuit/kit/rh/focus"
	"circuit/kit/rh/shutter"
	"circuit/kit/rh/xy"
	"circuit/use/circuit"
)

// Resources captures the local OS resources, shared with other workers
type Resources struct {
	Kin               *kinfolk.Kin
	KinJoin, KinLeave <-chan kinfolk.KinXID
	TubeTopic         string    // Topic of loci broadcast network
	FS                rh.Server // Shared local file system
}

// Locus
type Locus struct {
	Peer *focus.Peer
	//
	rsc        *Resources
	tube       *tube.Tube
	shutter    *shutter.Shutter
	shutterFID rh.FID // Non-cross
}

//
func NewLocus(rsc *Resources) *Locus {
	var err error
	//
	locus := &Locus{
		rsc:  rsc,
		tube: tube.New(rsc.Kin, rsc.TubeTopic),
	}
	// Create FID to local file system for shutter namespace
	var fsFID rh.FID
	if rsc.FS != nil {
		fsSSN, err := rsc.FS.SignIn("", "/")
		if err != nil {
			panic(err)
		}
		if fsFID, err = fsSSN.Walk(nil); err != nil {
			panic(err)
		}
	}
	// Create shutter server
	var shutterServer rh.Server
	if fsFID != nil {
		if locus.shutter, shutterServer, err = shutter.New(rsc.Kin.XID(), shutter.R{"fs", fsFID}); err != nil {
			panic(err)
		}
	} else if locus.shutter, shutterServer, err = shutter.New(rsc.Kin.XID()); err != nil {
		panic(err)
	}
	//
	locus.Peer = &focus.Peer{
		//
		// It is crucial to use permanent cross-references, and not "plain" ones within
		// values stored inside the tube table. If cross-references are used, they are managed by the
		// cross-garbage collection system and therefore connections to the underlying workers are maintained.
		Kin:     rsc.Kin.XID(),
		Shutter: circuit.PermRef(xy.XServer{shutterServer}),
	}
	// Create shutter FID for local use
	shutterSSN, err := shutterServer.SignIn("locus/"+locus.Peer.Key(), "/")
	if err != nil {
		panic(err)
	}
	if locus.shutterFID, err = shutterSSN.Walk(nil); err != nil {
		panic(err)
	}
	//
	go loopJoin(rsc.KinJoin)
	go locus.loopLeave(rsc.KinLeave)
	go locus.loopExpire()
	//
	log.Println(locus.Peer.Key())
	return locus
}

func (locus *Locus) Shutter() (*shutter.Shutter, rh.FID) {
	return locus.shutter, locus.shutterFID
}

func (locus *Locus) Peers() []*focus.Peer {
	rr := locus.tube.BulkRead()
	s := make([]*focus.Peer, len(rr))
	for i, r := range rr {
		s[i] = r.Value.(*focus.Peer)
	}
	return s
}

func (locus *Locus) loopExpire() {
	const GarbageDuration = time.Second * 3
	var rev tube.Rev
	for {
		rev++
		//log.Printf("WRITING (%s,%d,%v)", locus.info.Key(), rev, locus.info)
		locus.tube.Write(locus.Peer.Key(), rev, locus.Peer)
		//
		time.Sleep(GarbageDuration / 2)
		deadline := time.Now().Add(-GarbageDuration)
		for _, r := range locus.tube.BulkRead() {
			locus.tube.Forget(r.Key, 0, deadline)
		}
	}
}

func loopJoin(kinjoin <-chan kinfolk.KinXID) {
	// Discard join events
	for {
		kinXID, ok := <-kinjoin
		if !ok {
			panic("u")
		}
		log.Println("kin joined", kinXID.String())
	}
}

func (locus *Locus) loopLeave(kinleave <-chan kinfolk.KinXID) {
	for {
		kinXID, ok := <-kinleave
		if !ok {
			panic("u")
		}
		locus.denounce(kinXID)
	}
}

func (locus *Locus) denounce(kinXID kinfolk.KinXID) {
	peer := &focus.Peer{Kin: kinXID}
	log.Println("denouncing", peer.Key())
	r := locus.tube.Lookup(peer.Key())
	if r == nil {
		return
	}
	locus.tube.Scrub(peer.Key(), r.Rev, r.Updated)
}
